package net.wlm.realtime.new_job;

import com.alibaba.fastjson.JSON;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.funcation.HeatMapProcessAllWindowFunction;
import net.wlm.realtime.funcation.CarProcessAllWindowFunction;
import net.wlm.realtime.utils.JdbcUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


import java.time.Duration;

/**
 * 实时在线车辆数、平均电池健康度、车辆地理分布热力图
 */

public class RealtimeVehicleCntAvgBatHealHeatMap {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源
        DataStream<String> vehicleData = KafkaUtil.consumerKafka(env, "vehicle-data");
        // 数据清洗
        DataStream<VehicleData> vehicleStream = cleanData(vehicleData);
        // 设置水位线，乱序步长为5min,使用事件时间
        SingleOutputStreamOperator<VehicleData> vehicleDataSingleOutputStreamOperator = vehicleStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<VehicleData>forBoundedOutOfOrderness(Duration.ofMinutes(5))
                .withTimestampAssigner((event, timestamp) -> event.getTimestamp()));
        // vehicleStream.print();
        // 实时车辆统计  平均电池健康度
        DataStream<String> dataStream = vehicleDataSingleOutputStreamOperator
                // 设置窗口5min
                .windowAll(TumblingEventTimeWindows.of(Time.minutes(5)))
                .process(new CarProcessAllWindowFunction());
        //  countStream.print(); // 2025-06-06 21:05:00  2025-06-06 21:10:00  6 39.88
        JdbcUtil.sinkToClickhouseUpsert(dataStream,"insert into vehicle_real.vehicle_statistics values(?,?,?,?)");
        // 车辆地理分布热力图
        DataStream<String> heatMapStream = vehicleDataSingleOutputStreamOperator.
                windowAll(TumblingEventTimeWindows.of(Time.minutes(5)))
                .process(new HeatMapProcessAllWindowFunction());
        //    heatMapStream.print();
        JdbcUtil.sinkToClickhouseUpsert(heatMapStream,"insert into vehicle_real.geohash_cnt values(?,?)");
        env.execute();
    }


    private static DataStream<VehicleData> cleanData(DataStream<String> vehicleData) {
        DataStream<VehicleData> dataStream = vehicleData.filter(json -> {
                    try {
                        return json != null && JSON.isValid(json);
                    } catch (Exception e) {
                        return false;
                    }
                })
                // 将合法JSON字符串转换为VehicleData对象
                .map(json -> JSON.parseObject(json, VehicleData.class));
        return dataStream;
    }

}
